package ink.xiaobaibai.emqtt;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * @description: MQTT服务对象
 * @author: 小白白
 * @create: 2021-06-15
 **/

public class EMqttService {

    private String subTopic;
    private String pubTopic;
    /**
     * clientId必须唯一
     */
    private String clientId;
    private MyMqttCallback mqttCallback;
    private MqttClient client;
    private String brokerUrl;
    private String username;
    private String password;
    private boolean shutdown;

    EMqttService(String brokerUrl, String username, String password, String subTopic, String pubTopic, String clientId, MyMqttCallback mqttCallback) {
        this.subTopic = subTopic;
        this.pubTopic = pubTopic;
        this.clientId = clientId;
        this.mqttCallback = mqttCallback;
        this.brokerUrl = brokerUrl;
        this.username = username;
        this.password = password;
        this.init();
    }

    /**
     * 自动初始化
     */
    private void init() {
        String broker = this.brokerUrl;
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            this.client = new MqttClient(broker, this.clientId, persistence);
            this.client.setCallback(this.mqttCallback);
        } catch (MqttException e) {
            e.printStackTrace();
        }
        this.mqttCallback.seteMqttService(this);
    }

    /**
     * 开始运行
     */
    public void start() {
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(this.username);
        connOpts.setPassword(this.password.toCharArray());
        connOpts.setCleanSession(true);
        try {
            this.client.connect(connOpts);
            if (this.subTopic != null) {
                this.client.subscribe(this.subTopic);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 重启方法: 专供回调类调用
     */
    void restart() {
        this.init();
        this.start();
    }

    /**
     * 发送消息
     */
    public void pushMessage(String content) {
        int qos = 2;
        MqttMessage message = new MqttMessage(content.getBytes());
        message.setQos(qos);
        //保留消息,目标客户端没上线,会保留到其上线后发送,但是只能保留一条没有什么用
//        message.setRetained(true);
        try {
            this.client.publish(this.pubTopic, message);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 主动关闭
     */
    public void close() {
        if (this.client != null) {
            try {
                this.shutdown = true;
                this.client.disconnect();
                this.client.close();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

    public boolean isShutdown() {
        return this.shutdown;
    }

}
